// ****************************************************************************
// HushDB
//
// @File: HeapFile.h
//
// AUTHOR:  Justin Shen
//
// NOTES: 
//     This file implements heap file, including heap file page and heap file.
//
// ****************************************************************************

#ifndef HUSHDB_HEAPFILE_H
#define HUSHDB_HEAPFILE_H

#include "..\Common\BasicType.h"
#include "Page.h"
#include "BufferManager.h"
#include "AccessMethodInterface.h"

namespace Hush
{
    namespace Database
    {
        class SimpleHeap
        {
        private:
            struct Header
            {
                Header(PageId lastPageId)
                    : lastPageId(lastPageId)
                {
                }
                PageId lastPageId;
            };

            class Scan : public IScan
            {
            public:
                Scan(SimpleHeap* heap)
                    : heap(heap)
                {
                    DataPage* headerPage = heap->bufferManager->GetPageAs<DataPage>(heap->headerPageId);
                    RowId headerRowId = headerPage->GetFirstRow();
                    currentRowId = headerPage->GetNextRow(headerRowId);
                }

                virtual RowPtr GetNextRow()
                {
                    if (currentRowId.pageId == EndOfPage)
                    {
                        return RowPtr(0, RowPtr::EndOfRow);
                    }

                    RowId resultRowId = currentRowId;

                    // Move to next row
                    DataPage* page = heap->bufferManager->GetPageAs<DataPage>(currentRowId.pageId);
                    currentRowId = page->GetNextRow(currentRowId);

                    // Move to next page
                    if (currentRowId.slotId == EndOfSlot)
                    {
                        currentRowId.pageId = page->GetNextPageId();
                        if (currentRowId.pageId != EndOfPage)
                        {
                            currentRowId = page->GetFirstRow();
                        }
                    }
                }

            private:
                RowId currentRowId;
                SimpleHeap* heap;
            };
        public:
            SimpleHeap(BufferManager* bufferManager)
                : bufferManager(bufferManager)
            {                       
                DataPage* headerPage = bufferManager->AllocatePage<DataPage>();
                headerPageId = headerPage->GetPageId();
                Header header(headerPage->GetPageId());
                headerPage->InsertRow(RowPtr((Byte*)&header, sizeof(Header)));                
            }

            PageId GetHeaderPageId() { return this->headerPageId; }

            RowId InsertRow(const RowPtr& rowPtr);
        private:
            Header* GetHeader()
            {
                DataPage* headerPage = bufferManager->GetPageAs<DataPage>(headerPageId);
                return (Header*)(headerPage->GetRowPtr(headerPage->GetFirstRow()).data);
            }

            DataPage* GetLastPage()
            {
                PageId lastPageId = GetHeader()->lastPageId;
                DataPage* lastPage = bufferManager->GetPageAs<DataPage>(lastPageId);
                return lastPage;
            }

            void SetLastPage(PageId lastPageId)
            {
                GetHeader()->lastPageId = lastPageId;
            }

            PageId headerPageId;
            BufferManager* bufferManager;
        };

        class DirectoryPage : public DataPageImpl<DataPageData>
        {
        public:
            struct Entry
            {
                Entry(PageId pageId, Int16 freeSpace)
                    : pageId(pageId), freeSpace(freeSpace)
                { }

                PageId pageId;
                Int16 freeSpace;
            private:
                Int16 padding;
            };

        public:
            RowId AddEntry(PageId pageId, Int16 freeSpace)
            {                
                RowId rowId = this->AllocateRow(sizeof(DirectoryPage::Entry));
                DirectoryPage::Entry* entry = (DirectoryPage::Entry*)(&data[slots[rowId.slotId].offset]);
                entry->pageId = pageId;
                entry->freeSpace = freeSpace;   
                return rowId;
            }            
        };


        // Note:
        // - We assume there is no empty directory pages.
        // - A new heap contains 1 directory page which points to an empty data page.
        class Heap
        {
        public:
            class HeapScan : public IScan
            {
            public:
                HeapScan(Heap* heap)
                    : heap(heap)
                {
                    currentDirectoryRowId.pageId = heap->headerPageId;
                    PopulateRowInfo();
                }

                RowPtr GetNextRow()
                {
                    if (currentDataRowId.pageId == EndOfPage)
                    {
                        return RowPtr(0, RowPtr::EndOfRow);
                    }

                    DataPage* dataPage = heap->bufferManager->GetPageAs<DataPage>(currentDataRowId.pageId);
                    RowPtr resultRowPtr = dataPage->GetRowPtr(currentDataRowId);

                    // Try to get the next PageId and SlotId;
                    currentDataRowId.slotId = dataPage->GetNextValidSlotId(currentDataRowId.slotId);
                    if (currentDataRowId.slotId == EndOfSlot)
                    {
                        // We have returned all the rows on the current page. Need to go to next page.
                        DirectoryPage* headerPage = GetCurrentDirectoryPage();
                        currentDirectoryRowId.slotId = headerPage->GetNextValidSlotId(currentDirectoryRowId.slotId);
                        if (currentDirectoryRowId.slotId == EndOfSlot)
                        {
                            // We need to go to next directory page.
                            currentDirectoryRowId.pageId = headerPage->GetNextPageId();
                            if (currentDirectoryRowId.pageId == EndOfPage)
                            {
                                currentDataRowId.pageId = EndOfPage;
                            }
                            else
                            {
                                PopulateRowInfo();
                            }
                        }
                        else
                        {
                            DirectoryPage::Entry* dirEntry = (DirectoryPage::Entry*)(headerPage->GetRowPtr(currentDirectoryRowId.slotId).data);
                            DataPage* dataPage = heap->bufferManager->GetPageAs<DataPage>(dirEntry->pageId);
                            this->currentDataRowId = dataPage->GetFirstRow();
                        }
                    }

                    return resultRowPtr;
                }

            private:
                inline DirectoryPage* GetCurrentDirectoryPage()
                {
                    return heap->bufferManager->GetPageAs<DirectoryPage>(currentDirectoryRowId.pageId);
                }

                //---------------------------------------------------------------------
                // Method:  PopulateRowInfo
                //
                // Description:
                //   This method populates the slotId of currentDirecotryRowId and currentDataRowId
                //   Based on the pageId of currentDirecotryRowId
                //   The populated row is the first row on the first page of the directory page.
                //
                // Notes:
                //  - Assumes the current directory page is valid.
                inline void PopulateRowInfo()
                {
                    RowId resultRowId;

                    DirectoryPage* headerPage = GetCurrentDirectoryPage();
                    currentDirectoryRowId.slotId = headerPage->GetFirstSlotId();
                    
                    if (currentDirectoryRowId.slotId == EndOfSlot)
                    {
                        currentDataRowId.pageId = EndOfPage;
                        return;
                    }
                    else
                    {
                        // Get the page id of first data page                        
                        RowPtr rowPtr = headerPage->GetRowPtr(currentDirectoryRowId);
                        currentDataRowId.pageId = ((DirectoryPage::Entry*)(rowPtr.data))->pageId;

                        // Get the slot id of first slot
                        DataPage* dataPage = heap->bufferManager->GetPageAs<DataPage>(currentDataRowId.pageId);
                        currentDataRowId.slotId = dataPage->GetFirstSlotId();
                    }       
                }

            private:
                Heap* heap;

                RowId currentDirectoryRowId;
                RowId currentDataRowId;
            };

            Heap(BufferManager* bufferManager)
                : bufferManager(bufferManager)
            {
                DirectoryPage* headerPage = bufferManager->AllocatePage<DirectoryPage>();
                headerPageId = headerPage->GetPageId();

                DataPage* dataPage = bufferManager->AllocatePage<DataPage>();
                currentDataPageId = dataPage->GetPageId();

                headerPage->AddEntry(currentDataPageId, dataPage->GetAvailableSpace());
                currentDirectoryRowId = headerPage->GetFirstRow();

            }

            Heap(PageId headerPageId, BufferManager* bufferManager)
                : headerPageId(headerPageId), bufferManager(bufferManager)
            {
                // Get the first data page from header
                DirectoryPage* headerPage = bufferManager->GetPageAs<DirectoryPage>(headerPageId);
                currentDirectoryRowId = headerPage->GetFirstRow();
               
                RowPtr rowPtr = headerPage->GetRowPtr(currentDirectoryRowId.slotId);
                currentDataPageId = ((DirectoryPage::Entry*)(rowPtr.data))->pageId;                 
            }

            RowId InsertRow(const RowPtr& rowPtr);
            HeapScan* OpenScan() { return new HeapScan(this); }

        private:
            PageId headerPageId;
            RowId currentDirectoryRowId;
            PageId currentDataPageId;
            BufferManager* bufferManager;
        };


    }
}



#endif